package com.bdqn.spark.chapter06.bean

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Spark03_HotCategoryTop10SessionAnalysis {
  def main(args: Array[String]): Unit = {
    // TODO : Top10热门品类
    val sparConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("HotCategoryTop10Analysis")
    val sc = new SparkContext(sparConf)

    // todo：第一步：读取原始数据
    val actionRDD: RDD[String] = sc.textFile("input/user_visit_action.csv")

    actionRDD.cache()

    // 获取到前10个热门的品类id
    val top10CategoryIds: Array[String] = top10Category(actionRDD)

    // 1、数据中不包含这10个热门品类的数据，不要
    // 第一个是点击数据
    // 第二个，点击数据的品类肯定是在top10品类里
    val filterActionRDD: RDD[String] = actionRDD.filter(
      line => {
        val fields = line.split(",")
        if (fields(6) == "-1") {
          false
        } else {
          top10CategoryIds.contains(fields(6))
        }

      }
    )

    // （（session,category_id），1） ->
    val reduceRDD: RDD[((String, String), Int)] = filterActionRDD.map(
      line => {
        val fields = line.split(",")
        ((fields(2), fields(6)), 1)
      }
    ).reduceByKey(_ + _)

    // （category_id，（session,1））
    //    reduceRDD.map(
    //      tuple => {
    //        (tuple._1._2, (tuple._1._1, tuple._2))
    //      }
    //    )
    val mapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      // // （（session,category_id），1）
      case ((sid, cid), cnt) => {
        (cid, (sid, cnt))
      }
    }

    val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()

    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues {
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    }

    resultRDD.collect().foreach(println)


    sc.stop()
  }

  def top10Category(actionRDD: RDD[String]): Array[String] = {
    // TODO: 第二步：统计品类的点击总数：（品类，点击总数）
    val clickActionFilterRDD: RDD[String] = actionRDD.filter(
      // 2019-05-05,85,e2eef06e-beaa-4b49-acaf-38e057e1cd6e,36,2019-05-05 02:20:02,,-1,-1,,,1-2-3,1-2-3,8
      line => {
        val fields: Array[String] = line.split(",")
        fields(6) != "-1"
      }
    )

    val clickCategoryToOneRDD: RDD[(String, Int)] = clickActionFilterRDD.map(
      line => {
        val fields = line.split(",")
        (fields(6), 1)
      }
    )

    // 品类的点击总数 （品类，点击总数）
    val clickActionRDD: RDD[(String, Int)] =
      clickCategoryToOneRDD.reduceByKey(_ + _)

    // TODO: 第三步：统计品类的下单总数：（品类，下单总数）
    val orderActionFilterRDD: RDD[String] = actionRDD.filter(
      line => {
        val fileds: Array[String] = line.split(",")
        fileds(8) != ""
      }
    )

    // 一对多
    val orderCategoryToOneRDD: RDD[(String, Int)] = orderActionFilterRDD.flatMap(
      line => {
        val fields: Array[String] = line.split(",")
        val categoryIds: String = fields(8) // 1-2-3
        val arrIDs: Array[String] = categoryIds.split("-") // [1,2,3]
        arrIDs.map((_, 1)) // (1,1) (2,1) (3,1)
      }
    )

    // 品类的下单总数 （品类，下单总数）
    val orderActionRDD: RDD[(String, Int)] = orderCategoryToOneRDD.reduceByKey(_ + _)

    // TODO: 第四步：统计品类的支付总数：（品类，支付总数）
    val payActionFilterRDD: RDD[String] = actionRDD.filter(
      line => {
        val fileds: Array[String] = line.split(",")
        fileds(10) != ""
      }
    )

    // 一对多
    val payCategoryToOneRDD: RDD[(String, Int)] = payActionFilterRDD.flatMap(
      line => {
        val fields: Array[String] = line.split(",")
        val categoryIds: String = fields(10) // 1-2-3
        val arrIDs: Array[String] = categoryIds.split("-") // [1,2,3]
        arrIDs.map((_, 1)) // (1,1) (2,1) (3,1)
      }
    )

    // 品类的下单总数 （品类，下单总数）
    val payActionRDD: RDD[(String, Int)] = payCategoryToOneRDD.reduceByKey(_ + _)


    // TODO: （品类，点击总数） （品类，下单总数） （品类，支付总数）
    // TODO: （品类，（点击总数，下单总数，支付总数））
    // 一个是 把 三个rdd 连接在一起  connect
    // 另一个是 连完后，按照相同的品类，进行分组，group
    val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] =
    clickActionRDD.cogroup(orderActionRDD, payActionRDD)

    val analysisRDD: RDD[(String, (Int, Int, Int))] = cogroupRDD.mapValues {
      case (clickIte, orderIte, payIte) => {
        var clickCnt = 0
        val iterClick = clickIte.iterator
        if (iterClick.hasNext) {
          clickCnt = iterClick.next()
        }

        var orderCnt = 0
        val iterOrder: Iterator[Int] = orderIte.iterator
        if (iterOrder.hasNext) {
          orderCnt = iterOrder.next()
        }

        var payCnt = 0
        val iterPay: Iterator[Int] = payIte.iterator
        if (iterPay.hasNext) {
          payCnt = iterPay.next()
        }

        (clickCnt, orderCnt, payCnt)
      }
    }

    // 按照 元组的第二个元素（三元组）排序
    val result: Array[(String, (Int, Int, Int))] = analysisRDD.sortBy(_._2, false).take(10)
    result.map(_._1)
  }
}
